package com.dingzhen.config;

import com.dingzhen.utils.SystemUtils;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.handler.annotation.Header;

import javax.annotation.Resource;

@Configuration
@IntegrationComponentScan
public class MqttConfig {
    @Resource
    private MqttProperties mqttProperties;
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        PropertyMapper mapper = PropertyMapper.get();
        mapper.from(mqttProperties::getMqttServices).whenNonNull().to(options::setServerURIs);
        mapper.from(mqttProperties::getUser).whenNonNull().to(options::setUserName);
        mapper.from(mqttProperties::getPassword).whenNonNull().as(String::toCharArray).to(options::setPassword);
        mapper.from(mqttProperties::getKeepAliveInterval).whenNonNull().to(options::setKeepAliveInterval);
        mapper.from(mqttProperties::getAutomaticReconnect).whenNonNull().to(options::setAutomaticReconnect);
        mapper.from(mqttProperties::getCleanSession).whenNonNull().to(options::setCleanSession);
        mapper.from(mqttProperties::getCompletionTimeout).whenNonNull().to(options::setConnectionTimeout);
        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(
                        String.valueOf(SystemUtils.nextId()),
                        mqttClientFactory(),
                        "+"
//                        mqttProperties.getInputTopic()
                );
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(mqttProperties.getQos());
        adapter.setOutputChannel(mqttInputChannel());
        adapter.setCompletionTimeout(mqttProperties.getCompletionTimeout());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        System.err.println("检查入站消息设置！");
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                System.err.println("有新消息入站了！");
            }

        };
    }


    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        System.err.println("发送消息设置规则！");
        MqttPahoMessageHandler messageHandler =
                new MqttPahoMessageHandler(String.valueOf(SystemUtils.nextId()), mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setAsyncEvents(true);
        messageHandler.setDefaultTopic(String.valueOf(mqttProperties.getOutTopic()));
        System.err.println("发送消息设置规则完成！");
        return messageHandler;
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
    public interface MyGateway {
        // 向默认的 topic 发送消息
        void sendToMqtt(String payload);

        // 向指定的 topic 发送消息
        void sendToMqtt(String payload, @Header(MqttHeaders.TOPIC) String topic);

        // 向指定的 topic 发送消息，并指定服务质量参数
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);

        // 指定topic、qos、retained进行消息发送
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, @Header(MqttHeaders.RETAINED) boolean retained, String payload);
    }
}
